package com.com.zs21cp.watermark;

import com.com.zs21cp.ToolBean.ClickSource;
import com.com.zs21cp.ToolBean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class watremark_01 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource dataStreamSource = env.addSource(new ClickSource());



        WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy
            // WatermarkStrategy.forBoundedOutOfOrderness/noWatermarks/forMonotonousTimestamps
            //分别代表着乱序/无/有序
            //需要在有序乱序参数之前定义对象的类型，
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                @Override
                public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                    return element.getTs() * 1000L;
                }
            });
        //这是指定一个字段或者数据作为watermark的生成依据，其中后面的参数，
        //kafkasource中是不需要指定获取时间戳的
        SingleOutputStreamOperator singleOutputStreamOperator = dataStreamSource.assignTimestampsAndWatermarks(waterSensorWatermarkStrategy);


//        dataStreamSource.print();
        env.execute();
    }
}
